package com.fly.easy.queue.croe;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;

import com.fly.easy.queue.common.Channel;
import com.fly.easy.queue.common.ConsumerTask;
import com.fly.easy.queue.common.MessageEntity;
import com.fly.easy.queue.utils.ThreadUtil;

public class FlyClient {
	/**
	 * 日志
	 */
	private static Logger logger = Logger.getLogger(FlyClient.class);
	
	/**
	 * 注册handleClass
	 */
	private List<String>handleClassName=new ArrayList<>();
	
	
	/**
	 * this
	 */
	private FlyClient client=this;

	/**
	 * 消息频道
	 */
	private  Map<String, Channel> channels = new HashMap<>();
	
	/**
	 * 消费线程数量
	 */
	private  int consumer_thread_numer =5;
	
	/**
	 *  生产线程池数量
	 */
	private  int produce_thread_numer =5;
	


	/**
	 * 队列消费Handle
	 */
	private  Map<String, FlyMessageHandle> registerHandle = new HashMap<>();
	
	/**
	 * 生产线程池
	 */
	private ThreadPoolExecutor produce_executor = new ThreadPoolExecutor(produce_thread_numer, produce_thread_numer, 200, TimeUnit.MILLISECONDS,
			new ArrayBlockingQueue<Runnable>(1000));
	
	/**
	 * 消费线程池
	 */
	private ThreadPoolExecutor consumer_executor = new ThreadPoolExecutor(consumer_thread_numer, consumer_thread_numer, 200, TimeUnit.MILLISECONDS,
			new ArrayBlockingQueue<Runnable>(10));
	
	/**
	 * 初始化方法
	 */
	public void init() {
		//注册类
		for (String cl : handleClassName) {
			registerHandleClass(cl);
		}
		//开启消费
		consumer();
	}
	
	/**
	 * 销毁
	 */
	public void destroy() {
		produce_executor.shutdown();
		produce_executor.shutdown();
	}

	/**
	 * 注册处理类
	 * 
	 * @param name
	 */
	public  void registerHandleClass(String name) {
		try {
			String key = name.trim();
			Class<?> c = Class.forName(key);
			FlyMessageHandle handle = (FlyMessageHandle) c.newInstance();
			registerHandle.put(key, handle);
			channels.put(key, new Channel());
			logger.debug("register :".concat(name));
		} catch (ClassNotFoundException e) {
			throw new RuntimeException("Class Not Found check you config", e);
		} catch (InstantiationException | IllegalAccessException e) {
			throw new RuntimeException("create FlyMessageHandle error", e);
		}
	}

	/**
	 * 注册处理类
	 * 
	 * @param name
	 */
	public  FlyMessageHandle queryHandle(String name) {
		FlyMessageHandle handle = registerHandle.get(name.trim());
		if (handle == null) {
			throw new RuntimeException("Not Found FlyMessageHandle");
		}
		return handle;
	}

	
	
	/**
	 * 非阻塞发送
	 * 用线程池进行的发送，适合于批量发送，可能会发送错误需要自行处理
	 * @param entity
	 * @param timeout 等待尝试时间 （秒）
	 */
	public void sendMultithread(MessageEntity entity,long timeout) throws  RejectedExecutionException{
		Channel channel = channels.get(entity.getHandle());
		if (channel == null) {
			throw new RuntimeException("a useless channel");
		}
		try {
			produce_executor.execute(new Runnable() {
				public void run() {
					channel.getMessageQueue().push(entity);
				}
			});
		} catch (RejectedExecutionException e) {
			try {
				ThreadUtil.sleep(timeout);
				produce_executor.execute(new Runnable() {
					public void run() {
						channel.getMessageQueue().push(entity);
					}
				});
			} catch (Exception e2) {
				logger.error("RejectedExecutionException error",e2);
				throw e2;
			}
		}
	}
	
	
	/**
	 * 安全的发送（频道队列满返回false 加入不成功）
	 * @param entity
	 * @return
	 */
	public  boolean  sendOffer(MessageEntity entity) {
		Channel channel = channels.get(entity.getHandle());
		if (channel == null) {
			throw new RuntimeException("a useless channel");
		}
		return channel.getMessageQueue().offer(entity);
	}
	
	
	/**
	 * 安全的发送（频道队列满返回false 加入不成功）
	 * @param entity
	 * @param time out 尝试时间秒
	 * @return
	 */
	public  boolean  sendOffer(MessageEntity entity,long timeout) {
		Channel channel = channels.get(entity.getHandle());
		if (channel == null) {
			throw new RuntimeException("a useless channel");
		}
		return channel.getMessageQueue().offer(entity,timeout);
	}
	
	/**
	 * 安全的发送（频道队列满会阻塞程序）
	 * @param entity
	 * @return
	 */
	public boolean synSend(MessageEntity entity) {
		Channel channel = channels.get(entity.getHandle());
		if (channel == null) {
			throw new RuntimeException("a useless channel");
		}
		return channel.getMessageQueue().push(entity);
	}

	/**
	 * 订阅消费
	 */
	public  void consumer() {
		if(channels.size()>0) {
			consumer_executor.setCorePoolSize(consumer_thread_numer * channels.size());
			consumer_executor.setMaximumPoolSize(consumer_thread_numer * channels.size());
		}
		for (String key : channels.keySet()) {
			Channel ch = channels.get(key);
			for (int i = 0; i < consumer_thread_numer; i++) {
				consumer_executor.execute(new ConsumerTask(ch, client));
			}
		}

	}

	public List<String> getHandleClassName() {
		return handleClassName;
	}

	public void setHandleClassName(List<String> handleClassName) {
		this.handleClassName = handleClassName;
	}

	public int getConsumer_thread_numer() {
		return consumer_thread_numer;
	}

	public void setConsumer_thread_numer(int consumer_thread_numer) {
		this.consumer_thread_numer = consumer_thread_numer;
	}

	public int getProduce_thread_numer() {
		return produce_thread_numer;
	}

	public void setProduce_thread_numer(int produce_thread_numer) {
		this.produce_thread_numer = produce_thread_numer;
	}

}
